package SparkChat

import SparkChat.CommonInfo.{HearBeat, RemoveWorkInfo, ResignInfo, ResignResultMsg, ScheduleCheck, WorkInfo, startMsg}
import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

import scala.collection.mutable
import scala.concurrent.duration.DurationInt


class SparkMaster extends Actor {
  val workers = mutable.HashMap[String,WorkInfo]()
  override def receive: Receive = {
    case startMsg(msg)=>{
      println("SparkMaster 启动")
      self ! ScheduleCheck()
    }
    case ScheduleCheck()=>{
      import context.dispatcher
      println("启动定时检测心跳")
      context.system.scheduler.schedule(0 millis, 10000 millis, self, RemoveWorkInfo)
    }
    case RemoveWorkInfo=>{
      println("心跳检测")
      val currentTime = System.currentTimeMillis()
      val workInfo = workers.values

      workInfo.filter(currentTime - _.modifytime >6000).foreach(workInfo=> workers.remove(workInfo.id))
    }
    case HearBeat(id)=>{
      workers(id).modifytime = System.currentTimeMillis()
      println(s"$id work心跳时间已经更新")
    }
    case ResignInfo(id)=>{
      if(!workers.contains(id)){
        val workInfo = new WorkInfo(id)
        workers+=(id ->workInfo)
        println(s"$id work注册完成")
        sender ! ResignResultMsg(id)
      }
    }

  }
}

object SparkMaster extends App{
  val masterHost = "127.0.0.1"
  val masterPort = 10002
  val config = ConfigFactory.parseString(
    s"""
       |akka.actor.provider="akka.remote.RemoteActorRefProvider"
       |akka.remote.netty.tcp.hostname=$masterHost
       |akka.remote.netty.tcp.port=$masterPort
      """.stripMargin
  )
  val myfactory = ActorSystem("SparkMaster", config )
  val sparkMaster = myfactory.actorOf(Props[SparkMaster], "SparkMaster")
  sparkMaster ! startMsg(s"host:$masterHost port:$masterPort")
}
